package com.pg.agent.engine.service.utils;


import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.pg.agent.engine.client.dto.SOPCallVO;
import com.pg.agent.engine.client.message.InPutMessage;
import com.pg.agent.engine.client.model.llm.CallLog;
import com.pg.agent.engine.client.vo.HttpParam;
import com.pg.base.enums.ResultEnum;
import com.pg.base.exception.BizException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Response;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;

import javax.servlet.http.HttpServletResponse;
import java.util.concurrent.CountDownLatch;

@Slf4j
@Data
public class ConvertSSEListener extends EventSourceListener {

    private CountDownLatch countDownLatch = new CountDownLatch(1);

    private HttpParam httpParam;

    private String messageId;

    private HttpServletResponse rp;

    private SOPCallVO sopCallVO;


    private InPutMessage message;

    private boolean last;

    private boolean isFirst = true;

    private String jsonValue;

    private StringBuffer output;

    public ConvertSSEListener(HttpParam httpParam, InPutMessage message, HttpServletResponse response, String messageId, SOPCallVO sopCallVO, boolean last) {
        this.httpParam = httpParam;
        this.rp = response;
        this.message = message;
        this.messageId = messageId;
        this.sopCallVO = sopCallVO;
        this.last = last;
        this.output = new StringBuffer();
    }

    /**
     * {@inheritDoc}
     * 建立sse连接
     */
    @Override
    public void onOpen(final EventSource eventSource, final Response response) {
        if (rp != null) {
            rp.setContentType("text/event-stream");
            rp.setCharacterEncoding("UTF-8");
            rp.setStatus(200);
            log.info("建立sse连接..." + JSON.toJSONString(httpParam));
            sopCallVO.setTimeLong(System.currentTimeMillis() - sopCallVO.getCallStartTime().getTime());
            if (message != null && message.isDebug()) {
                sopCallVO.setInfoList(BeanUtil.copyToList(message.getCallLogList(), CallLog.class));
            }

        } else {
            log.info("客户端非sse推送" + JSON.toJSONString(httpParam));
        }
    }

    /**
     * 事件
     *
     * @param eventSource
     * @param id
     * @param type
     * @param data
     */
    @Override
    public void onEvent(EventSource eventSource, String id, String type, String data) {
        try {
            String content = null;
            try {
                if (isFirst) {
                    jsonValue = data;
                    isFirst = false;
                }
                log.info("data:{}", data);
                JSONObject result = JSONObject.parseObject(data);
                content = result.getJSONObject("data").getString("content");

            } catch (Exception e) {
                content = data;
            }
            if (StringUtils.isNotEmpty(content)) {
                output.append(content);
            }
        } catch (Exception e) {
            log.info("data:{}", data);
            log.error("消息错误[" + JSON.toJSONString(httpParam) + "]", e);
            countDownLatch.countDown();
            throw new BizException(ResultEnum.ERROR, e);
        }

    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onClosed(final EventSource eventSource) {
        //     log.info("sse连接关闭:{}", param.getMessageId());
        log.info("结果输出:{}", output.toString());
        countDownLatch.countDown();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onFailure(final EventSource eventSource, final Throwable t, final Response response) {
        //    log.error("使用事件源时出现异常... [响应：{}]...", param.getMessageId());
        countDownLatch.countDown();
    }

    public CountDownLatch getCountDownLatch() {
        return this.countDownLatch;
    }
}

